package co.yixiang.modules.shop.rest;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RestController;

import java.text.SimpleDateFormat;
import java.util.Date;

@Slf4j
@RestController
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class TDMQMessageController {
	//private final static String QUEUE_NAME = "orderCommission";

	/*
	 * public static void initTDMQ() { if(client ==null) { try { client =
	 * PulsarClient.builder().serviceUrl("pulsar://172.21.0.14:6000/")
	 * .listenerName("custom:pulsar-oonjr2vzdo/vpc-r9z7v0vc/subnet-6j2pevlp")
	 * .authentication(AuthenticationFactory.token(
	 * "eyJrZXlJZCI6InB1bHNhci1vb25qcjJ2emRvIiwiYWxnIjoiSFMyNTYifQ.eyJzdWIiOiJ3eF9yZWFkIn0.2MxN4K62sHouQXbLJXzIrNEvf45vBzPBiHlxh6YyvNo"
	 * )) .build(); } catch (PulsarClientException e) { // TODO Auto-generated catch
	 * block e.printStackTrace(); } } if(producer ==null) { try { producer =
	 * client.newProducer()
	 * .topic("persistent://pulsar-oonjr2vzdo/orderCommission/orderCommission_tdmq")
	 * .create(); } catch (PulsarClientException e) { // TODO Auto-generated catch
	 * block e.printStackTrace(); } } if(client1 ==null) { try { client1 =
	 * PulsarClient.builder().serviceUrl("pulsar://172.21.0.14:6000/")
	 * .listenerName("custom:pulsar-oonjr2vzdo/vpc-r9z7v0vc/subnet-6j2pevlp")
	 * .authentication(AuthenticationFactory.token(
	 * "eyJrZXlJZCI6InB1bHNhci1vb25qcjJ2emRvIiwiYWxnIjoiSFMyNTYifQ.eyJzdWIiOiJ3eF9jdXN0b20ifQ.IKE26NFA1HbWE8C_xw3cZTgiNge1wwV-G1eXITOr_aM"
	 * )) .build(); } catch (PulsarClientException e) { // TODO Auto-generated catch
	 * block e.printStackTrace(); } } if(consumer ==null) { try { consumer =
	 * client1.newConsumer()
	 * .topic("persistent://pulsar-oonjr2vzdo/orderCommission/orderCommission_tdmq")
	 * // .subscriptionType(SubscriptionType.Shared) //
	 * .enableRetry(true)默认关闭，如果需要重试则开启
	 * .subscriptionName("consumer005").subscribe(); } catch (PulsarClientException
	 * e) { // TODO Auto-generated catch block e.printStackTrace(); } } }
	 */

	public void tdmqMessage(String order_id) {
		PulsarClient client =null;
		PulsarClient client1 =null;
		Producer<byte[]> producer  =null;
		Consumer<byte[]> consumer  =null;
		SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 设置日期格式
		try {
			System.out.printf("进入TDMQ队列===========");
			log.info("进入TDMQ队列===========");
			//initTDMQ();
			
			 /* client = PulsarClient.builder().serviceUrl("pulsar://172.21.0.14:6000/")
			  .listenerName("custom:pulsar-oonjr2vzdo/vpc-r9z7v0vc/subnet-6j2pevlp")
			  .authentication(AuthenticationFactory.token(
			  "eyJrZXlJZCI6InB1bHNhci1vb25qcjJ2emRvIiwiYWxnIjoiSFMyNTYifQ.eyJzdWIiOiJ3eF9yZWFkIn0.2MxN4K62sHouQXbLJXzIrNEvf45vBzPBiHlxh6YyvNo"
			  )) .build();
			 
			
			  producer = client.newProducer()
			  .topic("persistent://pulsar-oonjr2vzdo/orderCommission/orderCommission_tdmq")
			  .create();*/
			 
			//测试空间
			  client = PulsarClient.builder().serviceUrl("http://pulsar-p3e9omdmeb.tdmq-jpnc.ap-gz.qcloud.tencenttdmq.com:5121")
			  //.listenerName("custom:pulsar-p3e9omdmeb/vpc-fs65a66n/subnet-jtagylkm")
			  .authentication(AuthenticationFactory.token(
			  "eyJrZXlJZCI6InB1bHNhci1wM2U5b21kbWViIiwiYWxnIjoiSFMyNTYifQ.eyJzdWIiOiJwdWxzYXItcDNlOW9tZG1lYl9qdXhpX3JlYWRlciJ9.LF3qvB-Oh3W9H4whUTtqkOhtaDVGFxMTfYUJTg2fyXU"
			  )) .build();
			  producer = client.newProducer()
			  //.topic("persistent://pulsar-wb7adj8xoqpe/orderCommission/orderCommission_msg").
					  .topic("persistent://pulsar-p3e9omdmeb/orderCommission_juxi/orderCommission_juximsg").
			  create();
			//System.out.printf("进入TDMQ队列111111111===========" + order_id);
			log.info("进入TDMQ队列111111111===========" + order_id);
			String s = order_id;
			producer.send(s.getBytes());
			log.info("关闭连接开始时间=========" + df.format(new Date()));
			producer.close();
			client.close();
			log.info("关闭连接结束时间=========" + df.format(new Date()));
			//System.out.printf("进入TDMQ队列2222222222===========" + producer);
			//log.debug("进入TDMQ队列2222222222===========" + producer);

		} catch (Exception e) {
			System.out.println("报错Message" + e.getMessage());
			log.error("报错Message" + e.getMessage());
			e.printStackTrace();
		}
		System.out.println("执行结束===============");
		log.info("执行结束===============");
	}

	public static void main(String[] argv) throws Exception {
		/*
		 * try {
		 * 
		 * 
		 * PulsarClient client =
		 * PulsarClient.builder().serviceUrl("pulsar://172.16.0.9:6000/")
		 * .listenerName("custom:pulsar-bkqqdzvx3z/vpc-9lez2cx9/subnet-ozhd1fdo")
		 * .authentication(AuthenticationFactory.token(
		 * "eyJrZXlJZCI6InB1bHNhci1ia3FxZHp2eDN6IiwiYWxnIjoiSFMyNTYifQ.eyJzdWIiOiJ3eF9jdXN0b20ifQ.ZDRhX2EnMWducHugHH2LWgAgd8DfZ8-xjWhQ3ojuEZU"
		 * )) .build(); String s ="111111"; Producer<byte[]> producer =
		 * client.newProducer()
		 * .topic("persistent://pulsar-bkqqdzvx3z/orderCommission/orderCommission_msg").
		 * create(); producer.send(s.getBytes());
		 * 
		 * Consumer consumer = client.newConsumer().topic(
		 * "persistent://pulsar-bkqqdzvx3z/orderCommission/orderCommission_msg") //
		 * .subscriptionType(SubscriptionType.Shared) //
		 * .enableRetry(true)默认关闭，如果需要重试则开启
		 * .subscriptionName("consumer001").subscribe(); while (true) { // 等待接收消息
		 * Message msg = consumer.receive(); try {
		 * System.out.printf("Message received: %s", new String(msg.getData())); //
		 * 消息ACK consumer.acknowledge(msg); } catch (Exception e) { // 处理出错 //
		 * enableRetry=true 才能使用重试方法 consumer.reconsumeLater(msg, 1000L,
		 * TimeUnit.MILLISECONDS);// 消息按照指定的延迟时间重试 //
		 * consumer.reconsumeLater(msg);按照延迟等级进行重试，多次重试默认自增延迟等级 //
		 * consumer.reconsumeLater(msg, 1);按照指定的延迟等级进行重试 // delayLevel =1 代表1s, //
		 * delayLevel =2 代表5s, //
		 * 默认延迟等级"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h" }
		 * CommissionMain.checkOrderGoodsclass(new String(msg.getData())); } } catch
		 * (Exception e) { System.out.printf("报错", e.getMessage()); }
		 */
	}

}
